package com.ylkj.bus.dispatcher.handler;

import com.alibaba.fastjson2.JSONObject;
import com.hwl.redis.RedisConfig;
import com.ylkj.bus.monitor.equipment.model.EquipmentAlarmMonitorLogEnty;
import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.MessageFormat;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;

/**
 * @Author: wengzx
 * @Date: 2025/4/28
 * @Desc:
 */
@Slf4j
public class DeviceMonitorDataHandlerEnd extends ProcessFunction<EquipmentAlarmMonitorLogEnty, String> {
    // 定义日期时间格式
    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    // 下线第1次更新
    private static final String BASE_UPDATE_OFFLINE_SQL1 =
            "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`, `event_value`, `alarm_time`, `blanking_tool`, `product_id`,`product_type`,`message_suggestion`,`alert_source`,`net_code`,`personnel_handler`,`handling_time`,`alert_description`,create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?,  ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`alarm_time` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN VALUES(`alarm_time`) ELSE `alarm_time` END, " +
                    "`event_value` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN VALUES(`event_value`) ELSE `event_value` END, " +
                    "`alarm_cover` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN VALUES(`alarm_cover`) ELSE `alarm_cover` END, " +
                    "`update_time` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN now() ELSE `update_time` END";

    // 下线第2次更新
    private static final String BASE_UPDATE_OFFLINE_SQL2 =
            "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`, `event_value`, `alarm_time`, `blanking_tool`, `product_id`,`product_type`,`message_suggestion`,`alert_source`,`net_code`,`personnel_handler`,`handling_time`,`alert_description`,create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?,  ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`alarm_time` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`alarm_time`) ELSE `alarm_time` END,  " +
                    "`event_value` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`event_value`) ELSE `event_value` END,  " +
                    "`update_time` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN now() ELSE `update_time` END";

    // 上线第1次更新
    private static final String BASE_UPDATE_ONLINE_SQL1 =
            "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`,   `blanking_tool`, `product_id`,`product_type`,`alert_source`,`net_code`,`handling_time`,create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?,  ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`alert_level` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`alert_level`) ELSE `alert_level` END,  " +
                    "`alert_dimensions` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`alert_dimensions`) ELSE `alert_dimensions` END,  " +
                    "`alarm_cover` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`alarm_cover`) ELSE `alarm_cover` END,  " +
                    "`handling_time` =  CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN VALUES(`handling_time`) ELSE `handling_time` END,  " +
                    "`update_time` = CASE WHEN (alarm_time < ? OR alarm_time IS NULL) THEN now() ELSE `update_time` END";

    // 上线第2次更新
    private static final String BASE_UPDATE_ONLINE_SQL2 =
            "INSERT INTO {0} (`device_code`, `alert_level`, `alert_dimensions`, `alarm_cover`, `event_type`,   `blanking_tool`, `product_id`,`product_type`,`alert_source`,`net_code`,`handling_time`,create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?,  ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`handling_time` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN VALUES(`handling_time`) ELSE `handling_time` END,  " +
                    "`update_time` = CASE WHEN (handling_time < ? OR handling_time IS NULL) THEN now() ELSE `update_time` END";

    // 数据库配置
    private final RdsConfig rdsConfig;
    private final String tableName1;

    private RdsUtils rdsUtils;
    // 格式化后的SQL语句
    private String baseOfflineSql1;
    private String baseOfflineSql2;
    private String baseOnlineSql1;
    private String baseOnlineSql2;

    // 死锁错误码定义
    private static final int DEADLOCK_ERROR_CODE = 1213;
    // 最大重试次数
    private static final int MAX_RETRY_TIMES = 3;

    public DeviceMonitorDataHandlerEnd(String rdsTableName, RdsConfig rdsConfig) {
        this.rdsConfig = rdsConfig;
        this.tableName1 = rdsTableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        rdsUtils = new RdsUtils(rdsConfig);
        // 初始化SQL模板
        baseOfflineSql1 = MessageFormat.format(BASE_UPDATE_OFFLINE_SQL1, tableName1);
        baseOfflineSql2 = MessageFormat.format(BASE_UPDATE_OFFLINE_SQL2, tableName1);
        baseOnlineSql1 = MessageFormat.format(BASE_UPDATE_ONLINE_SQL1, tableName1);
        baseOnlineSql2 = MessageFormat.format(BASE_UPDATE_ONLINE_SQL2, tableName1);
    }

    @Override
    public void processElement(EquipmentAlarmMonitorLogEnty values, ProcessFunction<EquipmentAlarmMonitorLogEnty, String>.Context ctx, Collector<String> out) throws Exception {
        if (values == null) {
            log.warn("收到空设备监控数据，跳过处理");
            return;
        }
        String deviceCode = values.getDeviceCode();
        String eventType = values.getEventType();
        String alarmCover = values.getAlarmCover();
        Date handlingTime = values.getHandlingTime();
        Date alarmTime = values.getAlarmTime();
        // 验证状态有效性
        if (!isValidStatus(alarmCover)) {
            log.warn("无效的设备状态值: {}", alarmCover);
            return;
        }
        int retryCount = 0;
        while (retryCount < MAX_RETRY_TIMES) {
            try (Connection connection = rdsUtils.getConnection()) {
                connection.setAutoCommit(false);
                boolean updateResult = false;
                int[] affectedRowsHolder = new int[1];
                if ("deviceOnline".equals(eventType)) {
                    updateResult = handleOnlineStatus(connection, deviceCode, values, affectedRowsHolder);
                } else if ("deviceOffline".equals(eventType)) {
                    updateResult = handleOfflineStatus(connection, deviceCode, values, affectedRowsHolder);
                }
                connection.commit();
                log.debug("设备[{}]状态更新成功，影响行数: {}", deviceCode, updateResult ? 1 : 0);
                return;
            } catch (SQLException e) {
                handleSQLException(e, retryCount);
                retryCount++;
            }
        }
        log.error("设备[{}]状态更新失败，超过最大重试次数", deviceCode);
    }

    /**
     * 处理设备上线状态更新
     * @param connection 数据库连接
     * @param deviceCode 设备编码
     * @param values 实体
     * @param affectedRowsHolder 影响行数持有者
     * @return 是否成功更新状态
     */
    private boolean handleOnlineStatus(Connection connection, String deviceCode,
                                       EquipmentAlarmMonitorLogEnty values, int[] affectedRowsHolder) throws SQLException {
        boolean firstUpdate = executeOnlineUpdate1(connection, deviceCode, values, affectedRowsHolder);
        if (!firstUpdate) {
            executeOnlineUpdate2(connection, deviceCode, values);
        }
        return firstUpdate;
    }

    /**
     * 执行上线状态第一次更新
     */
    private boolean executeOnlineUpdate1(Connection connection, String deviceCode,
                                         EquipmentAlarmMonitorLogEnty values, int[] affectedRowsHolder) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseOnlineSql1)) {
            setOnlineParameters(stmt, deviceCode, values);
            // 设置ON DUPLICATE KEY UPDATE部分的参数
            Timestamp handlingTimeStr = convertToSqlDate(values.getHandlingTime());
            stmt.setTimestamp(12, handlingTimeStr);
            stmt.setTimestamp(13, handlingTimeStr);
            stmt.setTimestamp(14, handlingTimeStr);
            stmt.setTimestamp(15, handlingTimeStr);
            stmt.setTimestamp(16, handlingTimeStr);
            affectedRowsHolder[0] = stmt.executeUpdate();
            return affectedRowsHolder[0] > 0;
        }
    }



    /**
     * 执行上线状态第二次更新
     */
    private void executeOnlineUpdate2(Connection connection, String deviceCode,
                                      EquipmentAlarmMonitorLogEnty values) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseOnlineSql2)) {
            setOnlineParameters(stmt, deviceCode, values);
            // 设置ON DUPLICATE KEY UPDATE部分的参数
            Timestamp handlingTimeStr = convertToSqlDate(values.getHandlingTime());
            stmt.setTimestamp(12, handlingTimeStr);
            stmt.setTimestamp(13, handlingTimeStr);
            stmt.executeUpdate();
        }
    }

    /**
     * 处理设备下线状态更新
     * @param connection 数据库连接
     * @param deviceCode 设备编码
     * @param values 实体
     * @param affectedRowsHolder 影响行数持有者
     * @return 是否成功更新状态
     */
    private boolean handleOfflineStatus(Connection connection, String deviceCode,
                                        EquipmentAlarmMonitorLogEnty values, int[] affectedRowsHolder) throws SQLException {
        boolean firstUpdate = executeOfflineUpdate1(connection, deviceCode, values, affectedRowsHolder);
        if (!firstUpdate) {
            executeOfflineUpdate2(connection, deviceCode, values);
        }
        return firstUpdate;
    }

    /**
     * 执行下线状态第一次更新
     */
    private boolean executeOfflineUpdate1(Connection connection, String deviceCode,
                                          EquipmentAlarmMonitorLogEnty values, int[] affectedRowsHolder) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseOfflineSql1)) {
            setOfflineParameters(stmt, deviceCode, values);
            // 设置ON DUPLICATE KEY UPDATE部分的参数
            Timestamp alarmTimeStr = convertToSqlDate(values.getAlarmTime());
            stmt.setTimestamp(17, alarmTimeStr);
            stmt.setTimestamp(18, alarmTimeStr);
            stmt.setTimestamp(19, alarmTimeStr);
            stmt.setTimestamp(20, alarmTimeStr);
            affectedRowsHolder[0] = stmt.executeUpdate();
            return affectedRowsHolder[0] > 0;
        }
    }

    /**
     * 执行下线状态第二次更新
     */
    private void executeOfflineUpdate2(Connection connection, String deviceCode,
                                       EquipmentAlarmMonitorLogEnty values) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseOfflineSql2)) {
            setOfflineParameters(stmt, deviceCode, values);
            // 设置ON DUPLICATE KEY UPDATE部分的参数
            Timestamp alarmTimeStr = convertToSqlDate(values.getAlarmTime());
            stmt.setTimestamp(17, alarmTimeStr);
            stmt.setTimestamp(18, alarmTimeStr);
            stmt.setTimestamp(19, alarmTimeStr);
            stmt.executeUpdate();
        }
    }

    /**
     * 设置上线参数
     */
    private void setOnlineParameters(PreparedStatement stmt, String deviceCode,
                                     EquipmentAlarmMonitorLogEnty values) throws SQLException {
        int paramIndex = 1;
        stmt.setString(paramIndex++, deviceCode);
        stmt.setString(paramIndex++, values.getAlertLevel());
        stmt.setString(paramIndex++, values.getAlertDimensions());
        stmt.setString(paramIndex++, values.getAlarmCover());
        stmt.setString(paramIndex++, "last_offline");
        stmt.setString(paramIndex++, values.getBlankingTool());
        stmt.setInt(paramIndex++, values.getProductId());
        stmt.setString(paramIndex++, values.getProductType());
        stmt.setString(paramIndex++, values.getAlertSource());
        stmt.setString(paramIndex++, values.getNetCode());
        stmt.setTimestamp(paramIndex++, convertToSqlDate(values.getHandlingTime()));
    }
    /**
     * 设置下线参数
     */
    private void setOfflineParameters(PreparedStatement stmt, String deviceCode,
                                      EquipmentAlarmMonitorLogEnty values) throws SQLException {
        int paramIndex = 1;
        stmt.setString(paramIndex++, deviceCode);
        stmt.setString(paramIndex++, values.getAlertLevel());
        stmt.setString(paramIndex++, values.getAlertDimensions());
        stmt.setString(paramIndex++, values.getAlarmCover());
        stmt.setString(paramIndex++, "last_offline");
        stmt.setString(paramIndex++, values.getEventValue());
        stmt.setTimestamp(paramIndex++, convertToSqlDate(values.getAlarmTime()));
        stmt.setString(paramIndex++, values.getBlankingTool());
        stmt.setInt(paramIndex++, values.getProductId());
        stmt.setString(paramIndex++, values.getProductType());
        stmt.setString(paramIndex++, values.getMessageSuggestion());
        stmt.setString(paramIndex++, values.getAlertSource());
        stmt.setString(paramIndex++, values.getNetCode());
        stmt.setString(paramIndex++, values.getPersonnelHandler());
        stmt.setTimestamp(paramIndex++, convertToSqlDate(values.getHandlingTime()));
        stmt.setString(paramIndex++, values.getAlertDescription());
    }


    /**
     * 将 java.util.Date 转换为 java.sql.Timestamp
     */
    private Timestamp convertToSqlDate(java.util.Date date) {
        if (date != null) {
            return new Timestamp(date.getTime());
        }
        return null;
    }

    /**
     * 验证状态值有效性
     */
    private boolean isValidStatus(String status) {
        return "1".equals(status) || "3".equals(status);
    }

    /**
     * 处理SQL异常
     */
    private void handleSQLException(SQLException e, int retryCount) throws SQLException {
        if (e.getErrorCode() == DEADLOCK_ERROR_CODE && retryCount < MAX_RETRY_TIMES) {
            log.warn("死锁检测，正在重试（第{}次）... 错误信息: {}", retryCount + 1, e.getMessage());
        } else {
            log.error("数据库操作失败，错误信息: {}", e.getMessage(), e);
            throw e;
        }
    }

    @Override
    public void close() throws Exception {
        if (rdsUtils != null) {
            rdsUtils.close();
        }
    }
}
